package com.mjf.spark.day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-重新分区
 *    -coalesce
 *       默认不执行shuffle，一般用于缩减分区
 *    -repartition
 *       底层调用coalesce，默认执行shuffle，一般用于增加分区
 */
object Spark10_Transformation_coalesce {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark10_Transformation_coalesce")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6), 3)

    println("=========重新分区之前=========")
    rdd.mapPartitionsWithIndex{
      (index, datas) => {
        println(index + "--->" + datas.mkString(","))
        datas
      }
    }.collect()

    // 缩减分区
//    val newRDD: RDD[Int] = rdd.coalesce(2)
    // 增加分区
    // 注意：默认情况下，如果使用coalesce增加分区是不起作用的。因为底层没有执行shuffle
//    val newRDD: RDD[Int] = rdd.coalesce(4)
    // 如果扩大分区 使用 repartition
    val newRDD: RDD[Int] = rdd.repartition(4)   // 相当于coalesce(4,true)

    println("=========重新分区之后=========")
    newRDD.mapPartitionsWithIndex{
      (index, datas) => {
        println(index + "--->" + datas.mkString(","))
        datas
      }
    }.collect()

    // 关闭连接
    sc. stop()

  }
}
